package com.cp.www.ws;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.kurento.client.IceCandidate;
import org.kurento.client.KurentoClient;
import org.kurento.client.MediaPipeline;
import org.kurento.client.PlayerEndpoint;
import org.kurento.client.WebRtcEndpoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import com.cp.www.ws.listener.CandidateFoundListener;
import com.cp.www.ws.listener.ErrorListener;
import com.cp.www.ws.listener.MediaStateChangedListener;
import com.cp.www.ws.listener.StreamListener;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;

/**
 * 播放业务逻辑处理
 */
public final class PlayHandler extends TextWebSocketHandler {

	@Autowired(required = false)
	private KurentoClient kurento;

	private Logger log = LoggerFactory.getLogger(PlayHandler.class);

	private Gson gson = new GsonBuilder().create();

	private Map<String, UserSession> users = new ConcurrentHashMap<>();

	@Override
	public void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
		if (kurento == null) {
			sendError(session, "流媒体服务器初始化失败!");
			return;
		}
		JsonObject jsonMessage = gson.fromJson(message.getPayload(), JsonObject.class);
		String sessionId = session.getId();
		try {
			switch (jsonMessage.get("id").getAsString()) {
			case "start":
				start(session, jsonMessage);
				break;
			case "stop":
				stop(sessionId);
				break;
			case "onIceCandidate":
				onIceCandidate(sessionId, jsonMessage);
				break;
			default:
				sendError(session, "Invalid message with id " + jsonMessage.get("id").getAsString());
				break;
			}
		} catch (Throwable t) {
			log.error("Exception handling message {} in sessionId {}", jsonMessage, sessionId, t.getMessage());
			sendError(session, t.getMessage());
		}
	}

	private void start(final WebSocketSession session, final JsonObject jsonMessage) {
		int delay = 0;
		JsonElement delayel = jsonMessage.get("delay");
		if (delayel != null) {
			delay = delayel.getAsInt();
		}
		String rtspUrl = jsonMessage.get("rtspUrl").getAsString();
		UserSession oldUserSession = selectByKey(rtspUrl, delay);
		if (oldUserSession == null) {// 第一次播放
			playNewRtsp(session, jsonMessage, rtspUrl, delay);
		} else {// 复用播放，避免多次解码带来的cpu消耗
			playReuseRtsp(oldUserSession, session, jsonMessage, rtspUrl, delay);
		}
	}

	private void playNewRtsp(final WebSocketSession session, final JsonObject jsonMessage, final String rtspUrl,
			final int delay) {
		final UserSession user = new UserSession();
		MediaPipeline pipeline = kurento.createMediaPipeline();
		user.setMediaPipeline(pipeline);
		WebRtcEndpoint webRtcEndpoint = new WebRtcEndpoint.Builder(pipeline).build();
		user.setWebRtcEndpoint(webRtcEndpoint);
		PlayerEndpoint playerEndpoint = new PlayerEndpoint.Builder(pipeline, rtspUrl).withNetworkCache(delay).build();
		user.setPlayerEndpoint(playerEndpoint);
		user.setRtspUrl(rtspUrl).setDelay(delay).setPresenter(true).setWebSocketSession(session);
		users.put(session.getId(), user);
		playerEndpoint.connect(webRtcEndpoint);
		webRtcEndpoint.addIceCandidateFoundListener(new CandidateFoundListener(session));
		String sdpOffer = jsonMessage.get("sdpOffer").getAsString();
		String sdpAnswer = webRtcEndpoint.processOffer(sdpOffer);
		JsonObject response = new JsonObject();
		response.addProperty("id", "startResponse");
		response.addProperty("sdpAnswer", sdpAnswer);
		sendMessage(session, response.toString());
		webRtcEndpoint.addMediaStateChangedListener(new MediaStateChangedListener());
		webRtcEndpoint.gatherCandidates();
		playerEndpoint.addErrorListener(new ErrorListener(session));
		playerEndpoint.addEndOfStreamListener(new StreamListener(session));
		playerEndpoint.play();
		log.info("RTSP New Play start {}.", session.getId());
	}

	private void playReuseRtsp(final UserSession oldUserSession, final WebSocketSession session,
			final JsonObject jsonMessage, String rtspUrl, int delay) {
		final UserSession user = new UserSession();
		MediaPipeline pipeline = oldUserSession.getMediaPipeline();
		user.setMediaPipeline(pipeline);
		WebRtcEndpoint webRtcEndpoint = new WebRtcEndpoint.Builder(pipeline).build();
		user.setWebRtcEndpoint(webRtcEndpoint);
		PlayerEndpoint playerEndpoint = oldUserSession.getPlayerEndpoint();
		user.setPlayerEndpoint(playerEndpoint);
		user.setRtspUrl(rtspUrl).setDelay(delay).setWebSocketSession(session);
		users.put(session.getId(), user);
		playerEndpoint.connect(webRtcEndpoint);
		webRtcEndpoint.addIceCandidateFoundListener(new CandidateFoundListener(session));
		String sdpOffer = jsonMessage.get("sdpOffer").getAsString();
		String sdpAnswer = webRtcEndpoint.processOffer(sdpOffer);
		JsonObject response = new JsonObject();
		response.addProperty("id", "startResponse");
		response.addProperty("sdpAnswer", sdpAnswer);
		sendMessage(session, response.toString());
		webRtcEndpoint.addMediaStateChangedListener(new MediaStateChangedListener());
		webRtcEndpoint.gatherCandidates();
		log.info("RTSP Reuse Play start {}.", session.getId());
	}

	private UserSession selectByKey(String rtspUrl, int delay) {
		Set<String> keySet = users.keySet();
		for (String key : keySet) {
			UserSession session = users.get(key);
			if (session.isPresenter() && session.getRtspUrl().equals(rtspUrl) && session.getDelay() == delay) {
				return session;
			}
		}
		return null;
	}

	private void stop(String sessionId) {
		UserSession user = users.remove(sessionId);
		if (user != null) {
			log.info("RTSP play remove {}.", sessionId);
			if (user.isPresenter()) {
				if (hasOtherPlaying(user)) {
					user.setClose(true);// 标记为关闭状态
					users.put(sessionId, user);// 还原回去
				} else {
					user.release();
					log.info("RTSP play release {}.", sessionId);
				}
			} else {
				UserSession presenter = isTheLastOne(user);
				if (presenter != null) {
					users.remove(presenter.getWebSocketSession().getId());
					presenter.release();
					log.info("RTSP play release {}.", sessionId);
				}
			}
		}
	}

	private boolean hasOtherPlaying(UserSession user) {
		String rtspUrl = user.getRtspUrl();
		int delay = user.getDelay();
		Set<String> keySet = users.keySet();
		for (String key : keySet) {
			UserSession session = users.get(key);
			if (!session.isPresenter() && session.getRtspUrl().equals(rtspUrl) && delay == session.getDelay()) {
				return true;
			}
		}
		return false;
	}

	private UserSession isTheLastOne(UserSession user) {
		String rtspUrl = user.getRtspUrl();
		int delay = user.getDelay();
		Set<String> keySet = users.keySet();
		UserSession presenter = null;
		for (String key : keySet) {
			UserSession session = users.get(key);
			if (session.getRtspUrl().equals(rtspUrl) && delay == session.getDelay()) {
				if (!session.isPresenter()) {
					return null;
				} else {
					presenter = session;
				}
			}
		}
		if (presenter == null) {
			return null;
		}
		if (presenter.isClose()) {
			return presenter;
		}
		return null;
	}

	private void onIceCandidate(String sessionId, JsonObject jsonMessage) {
		UserSession user = users.get(sessionId);
		if (user != null) {
			JsonObject jsonCandidate = jsonMessage.get("candidate").getAsJsonObject();
			IceCandidate candidate = new IceCandidate(jsonCandidate.get("candidate").getAsString(),
					jsonCandidate.get("sdpMid").getAsString(), jsonCandidate.get("sdpMLineIndex").getAsInt());
			user.getWebRtcEndpoint().addIceCandidate(candidate);
		}
	}

	private void sendError(WebSocketSession session, String message) {
		if (users.containsKey(session.getId())) {
			JsonObject response = new JsonObject();
			response.addProperty("id", "error");
			response.addProperty("message", message);
			sendMessage(session, response.toString());
		}
	}

	private void sendMessage(WebSocketSession session, String message) {
		synchronized (session) {
			try {
				session.sendMessage(new TextMessage(message));
			} catch (Exception e) {
				log.error("Exception sending message:{}!", e.getMessage());
			}
		}
	}

	@Override
	public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
		stop(session.getId());
	}

}
